﻿// Copyright (c) .NET Core Community. All rights reserved.
// Licensed under the MIT License. See License.txt in the project root for license information.

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Runtime.Remoting.Contexts;
using System.Threading;
using DotNetFramework.CAP.Core.Logger;
using MQTTnet;
using MQTTnet.Client;
using MQTTnet.Serializer;

namespace DotNetFramework.CAP.Mqtt
{
    public class ConnectionPool : IConnectionPool, IDisposable
    {
        private readonly MqttOptions _options;
        private readonly ConcurrentQueue<IMqttClient> _mqttClientPool;
        private int _pCount;
        private int _maxSize;

        public ConnectionPool(ILogger logger,
            CapOptions capOptions,
            MqttOptions options)
        {
            _options = options;
            _mqttClientPool = new ConcurrentQueue<IMqttClient>();
            _maxSize = _options.ConnectionPoolSize;

            logger.LogDebug("CAP Mqtt servers: {0}", _options.Server);
        }

        public string ServersAddress => _options.Server;

        public IMqttClient RentProducer()
        {
            IMqttClient result;
            if (this._mqttClientPool.TryDequeue(out result))
            {
                Interlocked.Decrement(ref this._pCount);
                return result;
            }
            MqttClientOptionsBuilder mqttClientOptionsBuilder = new MqttClientOptionsBuilder();
            mqttClientOptionsBuilder.WithTcpServer(this._options.Server, new int?(this._options.Port));
            if (!string.IsNullOrEmpty(this._options.UserName) && !string.IsNullOrEmpty(this._options.Password))
            {
                mqttClientOptionsBuilder.WithCredentials(this._options.UserName, this._options.Password);
            }
            if (!string.IsNullOrEmpty(this._options.ClientId))
                mqttClientOptionsBuilder.WithClientId(this._options.ClientId);
            mqttClientOptionsBuilder.WithKeepAlivePeriod(TimeSpan.FromSeconds(2000.0));
            mqttClientOptionsBuilder.WithCleanSession(true);
            mqttClientOptionsBuilder.WithProtocolVersion(MqttProtocolVersion.V311);
            //mqttClientOptionsBuilder.WithCommunicationTimeout(TimeSpan.FromHours(1.0));
            IMqttClientOptions options = mqttClientOptionsBuilder.Build();
            result = this.BuildMqttClient(options);
            return result;
        }

        public bool Return(IMqttClient context)
        {
            if (Interlocked.Increment(ref this._pCount) <= this._maxSize)
            {
                this._mqttClientPool.Enqueue(context);
                return true;
            }
            context?.Dispose();
            Interlocked.Decrement(ref this._pCount);
            return false;
        }

        public void Dispose()
        {
            _maxSize = 0;

            while (_mqttClientPool.TryDequeue(out var context))
            {
                context.Dispose();

            }
        }
        protected virtual IMqttClient BuildMqttClient(IMqttClientOptions options)
        {
            IMqttClient client = new MqttFactory().CreateMqttClient();
            client.ConnectAsync(options).Wait();//.GetAwaiter().GetResult();
            return client;
        }
    }
}